package com.dj.ms.event.support;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * <br>
 * <p>异步事件处理器</p>
 *
 * <br>
 * @author ZJG
 * @version 1.0
 * @date 2021/7/26 下午6:07
 */
@Slf4j
public class AsyncEventBusExecutor extends ThreadPoolExecutor {

    private final ThreadLocal<Long> startTime = new ThreadLocal<>();

    public AsyncEventBusExecutor(int corePoolSize, int maximumPoolSize){
        super(corePoolSize, maximumPoolSize
                ,60
                , TimeUnit.SECONDS
                ,new LinkedBlockingQueue<>(1000)
                ,new ThreadFactoryBuilder().setNameFormat("asyncEvent-th-%d").build()
                ,new CallerRunsPolicy()
        );
    }

    public AsyncEventBusExecutor(int corePoolSize, int maximumPoolSize, int queueSize, String threadNamePrefix, boolean rejectedThrowException){
        super(corePoolSize, maximumPoolSize
                ,60
                , TimeUnit.SECONDS
                ,new LinkedBlockingQueue<Runnable>(queueSize)
                ,new ThreadFactoryBuilder().setNameFormat(threadNamePrefix.concat("-th-%d")).build()
                ,rejectedThrowException ? new AbortPolicy() : new CallerRunsPolicy()
        );
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        startTime.set(System.currentTimeMillis());
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        long costMillis = System.currentTimeMillis() - startTime.get();
        startTime.remove();
        int queueSize = this.getQueue().size();
        if(this.getPoolSize() == this.getMaximumPoolSize() && queueSize > 1){
            log.error("异步事件监听执行耗时毫秒：{}，当前线程数：{}（Core:{} Max:{}），剩余任务数：{}（Active:{} Completed:{} All:{}）"
                    , costMillis, this.getPoolSize(), this.getCorePoolSize(), this.getMaximumPoolSize()
                    , queueSize, this.getActiveCount(), this.getCompletedTaskCount(), this.getTaskCount());
        }else if(costMillis > 1000 || queueSize > 1){
            log.warn("异步事件监听执行耗时毫秒：{}，当前线程数：{}（Core:{} Max:{}），剩余任务数：{}（Active:{} Completed:{} All:{}）"
                    , costMillis, this.getPoolSize(), this.getCorePoolSize(), this.getMaximumPoolSize()
                    , queueSize, this.getActiveCount(), this.getCompletedTaskCount(), this.getTaskCount());
        }else if(costMillis > 100) {
            log.info("异步事件监听执行耗时毫秒：{}，当前线程数：{}（Core:{} Max:{}），剩余任务数：{}（Active:{} Completed:{} All:{}）"
                    , costMillis, this.getPoolSize(), this.getCorePoolSize(), this.getMaximumPoolSize()
                    , queueSize, this.getActiveCount(), this.getCompletedTaskCount(), this.getTaskCount());
        }else if(log.isDebugEnabled()){
            log.debug("异步事件监听执行耗时毫秒：{}，当前线程数：{}（Core:{} Max:{}），剩余任务数：{}（Active:{} Completed:{} All:{}）"
                    , costMillis, this.getPoolSize(), this.getCorePoolSize(), this.getMaximumPoolSize()
                    , queueSize, this.getActiveCount(), this.getCompletedTaskCount(), this.getTaskCount());
        }
    }
}
